 /*
  *@(#)RequestCtlImpl.java		7 Sept. 2000
  *
  * Copyright: MetaPlus Inc & Inquiry Management Systems Ltd.  2000
  *
  * This software is the proprietary information of MetaPlus Inc & 
  * Inquiry Management Systems Ltd.  
  * Use is subject to license terms.
  */
package com.ims.iaa;

import java.rmi.RemoteException;
import java.util.*;
import java.rmi.server.*;
import java.sql.Time;

/**
  * The <code>RequestCtlImpl</code> object represents the "Request Control 
  * Center" service. It implemented the <code>RequestCtlImpl</code> remote
  * interface to suppport services required by other objects. There should  
  * be only one instance exist in the application as it is used to centrally
  * control all the request objects ( <code>AdTrackRequest</code> objects). 
  *
  * @see RequestCtl
  *
  * @author Anthony Zhang
  *
  * @version 1.00, 25 Sept. 2000
  * @version 1.10, 26 Nov. 2004 by ShengMeng Bai
  *
  */
public class RequestCtlImpl extends UnicastRemoteObject implements RequestCtl {

  static final long FECTH_FREQUENCY = 200000; //fecth the database every 200 second 
  static final long DISPATCH_MONITOR_FREQUENCY = 10000; //monitor dispatch even every 10 seconds.
  static final long TEN_MINUTES = 600000;//10 minute
  /** Ad analyse requests waiting for processing    */	
  private LinkedList adRequests;
  
  private Vector dispatchedRequests;
  
  private Vector processedRequests;
  
  private SpiderCtl spiderCenter;
  
  private String hostName = "localhost";
  private int registryPort = 1099;   
  
  private Thread checkRequestThread;
  private Thread dispatchThread; 
  //++**************************Add by ShengMeng Bai at 2004-07-27
  /**Used to setting the sign of dispatch(m_bDispatch),*/ 
  private Thread m_objCrawlTimeThread;
  /**if m_bDispatch turn to false from true, checkRequestThread will not get request form database,
   * and <code>dispatchRequest()</code> will not dispatch new request.
   * @see dispatchRequest
   * @see startRequestThread
   */ 
  private volatile boolean m_bDispatch = true;
  // used to count the thread run times.
  private int m_iMonitorTimes = 0;
  //--**************************Add by ShengMeng Bai at 2004-07-27

  /** 
    *  
    */
 public RequestCtlImpl(int serverPort)      throws java.rmi.RemoteException
 {
 	if (IaaDebug.bAppointCrawlTime){
 		setCrawlTimeThread();
 	}
  	//++Add by ShengMeng Bai at 2004-10-04
  	hostName = IaaDebug.ServerHost;
  	//--Add by ShengMeng Bai at 2004-10-04
 	registryPort = ((serverPort == 0)? registryPort: serverPort);
  	adRequests = new LinkedList();
	dispatchedRequests = new Vector();
	processedRequests = new Vector();
  }
  
  /** 
    *  Gets ad track request from the Request Control Center
    */
  protected synchronized AdTrackRequest getAdTrackRequest() {
  
		 AdTrackRequest adReq = null; 
		 
				
  		 if( !adRequests.isEmpty()) {
  		 	adReq = (AdTrackRequest)adRequests.getFirst();
  		 }
	 	 try  {
		 	if (IaaDebug.Debug)  {
				LocalRequest lReq = adReq.getLocalRequest();
		 		log("get " + lReq );		
		 	}
		    removeAdTrackRequest(adReq);
			dispatchedRequests.addElement(adReq);
	 	 } catch (Exception e)  {
	 		log("" + e );	
	 	 }		 
  		 return adReq;
  }  	
  
  /** 
    *  Adds ad track request to the Request Control Center
    *
    *  @param adReq The ad track request which is waiting for proecess.    
    */
  public synchronized boolean addAdTrackRequest(LocalRequest lr)
        throws RemoteException {
	  
      if(IaaDebug.processDebug)
      {
      int count=TestUtil.count++;
      String tName=Thread.currentThread().getName();
      //Method A call Method B
      StackTraceElement[] st=new Exception().getStackTrace();
      StackTraceElement aA=st[1];
      StackTraceElement aB=st[0];
      String say=aA.getClassName()+":"+aA.getMethodName()+":"+aA.getLineNumber()+
      " call "+aB.getClassName()+":"+aB.getMethodName()+aB.getLineNumber();
      IaaUtil.log("wukun:"+count+":"+tName+":"+say);
      }
	  
		AdTrackRequest adReq = null;
		
		try  {
			adReq = new AdTrackRequestImpl(lr,registryPort);
		} catch (Exception e)  {
			log("error create request " + e);
			return false;
		}
		
		if( !adRequests.contains(adReq)) {
			adRequests.addLast(adReq);
		} 
		log("add request " + lr );			
		raiseDispatchEvent();
		return true;
  }
  
  /** 
    *  This method is only used when failed to dispatch the request to the spiders control center,
    *   so the request is added as the first one
    */
  private synchronized void addAdTrackRequest(AdTrackRequest adReq)
        throws RemoteException {
	
		if(IaaDebug.Debug) 
				log("add request to the Ads Request Queue again" );		
		if( !adRequests.contains(adReq)) {
			adRequests.addFirst(adReq);
		} 
		
		raiseDispatchEvent();
  }
  
  /** 
    *  Removes ad track request from the Request Control Center
    *
    *  @param adReq The ad track request which should be removed.
    */
  public synchronized boolean removeAdTrackRequest(AdTrackRequest adReq)
        throws RemoteException {
		if(IaaDebug.Debug)log("remove request from the Request Queue" );		
		return adRequests.remove(adReq);
  }

  /** 
    *  
    */
  protected void dispatchRequest(){  
  	    boolean ok_flag = true;
		
	    if(IaaDebug.Debug) log("try to connect Spider Control Center" );
					
  	    if ( spiderCenter == null) {
				spiderCenter = IaaUtil.getSpiderCenter(hostName,registryPort);	
				if ( spiderCenter == null) return;
  	    }	
		
		try {  
		 	if(IaaDebug.Debug) 
	  			log("ask Spdier Control Center if any spider available" );		
	  	  	while( spiderCenter.isReady() &&  ( !adRequests.isEmpty() )  ) {
			
	  	  		if ( !m_bDispatch ){
					log("Crawl time is over, request control center will begin work in next work day." );	
	  	  			break;
	  	  		}
				if(IaaDebug.Debug) 
					log("send request to the Spider Control Center" );	
						
			  	AdTrackRequest adReq = getAdTrackRequest();
			
				if (adReq != null) {
					if (!spiderCenter.processRequest(adReq) ) {
						if(IaaDebug.Debug) 
							log("send ads request to the Spider Control Center error " 
								+ adReq.getLocalRequest());
						//put the request back to the request queue, this time it should appear as the first one	
						addAdTrackRequest(adReq);
						dispatchedRequests.remove(adReq);
					}
				}
				
			}
		 	if(IaaDebug.Debug) 
	  			log("spider Queue empty or Request Queue empty" );
	  	 } catch (Exception e) {  
			 	log("communication with Spider Control Center error:" + e );
	  	 }
		 
  }
  
  public void finishAdRequest(AdTrackRequest adReq)	  {
		dispatchedRequests.remove(adReq);
		//do not add finished request into this vector, otherwise it will be too big if 
		//server never stop
		//processedRequests.addElement(adReq);
  }
  
  /** 
    *  
    */
  public void raiseDispatchEvent() {
	if(IaaDebug.Debug)log("Raise dispatch event." );
	if (dispatchThread==null) setDispatchThread();
	IaaDebug.assert_6(dispatchThread!=null);
	if(dispatchThread.isAlive()) {
		if(IaaDebug.Debug)log("dispatchThread still alive." );
	} else {
		if(IaaDebug.Debug)log("dispatchThread dead." );
		setDispatchThread();
		dispatchThread.start(); 
	}
  }

  private void setDispatchThread() { 
		dispatchThread = new IaaThread("RequestCtrl_Dispatch")  {
		  	public void run()  {
			  log("notified to start dispatch process." );	
			  // let the check request thread to get requests
			  if(IaaDebug.StoreDB)checkRequestThread.interrupt();
			  dispatchRequest();
			  log("finished the dispatch process." );	
			} // end of run()
	    } ; //end of anonymous class
  }		
  
  public Object[] inquiryRequest(byte state) {
  
 		switch (state)  {
			case AdTrackRequest.WAITING: 
				return adRequests.toArray();
			case AdTrackRequest.PROCESSING: 
				return dispatchedRequests.toArray();			
			case AdTrackRequest.PROCESSED: 
				return processedRequests.toArray();
			default: 
				return  new AdTrackRequest[1];
 		}
  }
  public void setRequestThread(Thread t)  {
  		checkRequestThread = t;
  }
  
  public void startRequestThread()  {
		checkRequestThread = new IaaThread("RequestCtrl_RequestLoad")  {
			SpiderDBConnection sp = 
				IaaUtil.getSpiderDBConnection(hostName,registryPort);
			LinkedList list = new LinkedList();
		
			public void run()  {
				if(IaaDebug.Debug)log("start thread to get request");		
				while (true)  {
					if (m_bDispatch){
						try {
							if(IaaDebug.Debug)log("start get request from database");		
							if ( sp == null)  {
								log("can not find the Spider DB Connection" );
								return;
							}
							//++changsheng 2006-11-8
			       			list = sp.loadCustomerRequest(IaaThread.getLogString());
							//--
							for (int i = 0;i<list.size(); i++)  {
								addAdTrackRequest((LocalRequest)(list.get(i)));
							}					
	    	   				if(IaaDebug.Debug)log("end get request from database" );
							//in case the dispatch thread did not succeffully dispatch all the requests
							//here we raise the dispatch event again  
							//if( spiderCenter.isReady() &&  ( !adRequests.isEmpty()) )raiseDispatchEvent();
							//Anthony zhang 14/Dec/2000 comment the following line, replace it with the above line.
							// Reason: If there are requests in the request center and no spiders free, once this 
							//thread notify the dispatch thread, it will become a infinite loop until all the 
							// requests are finished. 
							//The dispatch thread will ask this thread to check the database and 
							// this thread will ask the dispatch thread to dispatch.
							//if( !adRequests.isEmpty() )raiseDispatchEvent();						
				      	} catch ( RemoteException re ) {
				        		log(re.toString() );
	      				}
					}
					try  {
						Thread.sleep(FECTH_FREQUENCY);
					} catch (InterruptedException e)  {
						if(IaaDebug.Debug)log("interrupted by others");
					}
				}//end of while
			} // end of run()
			private void log(String s)  {
				IaaUtil.log("[Ads Request Thread] " + s);
			}
		} ; //end of anonymous class

		checkRequestThread.start();
  }
  
  private void log(String s)  {
	  IaaUtil.log( "[Request Control Center] " + s);
  }
  
  private void setCrawlTimeThread(){
  	m_objCrawlTimeThread = new IaaThread("RequestCtrl_CrawlTimeMonitor"){
  		public void run(){
  			while(true){
  				Time timeCurrent = new Time(System.currentTimeMillis());
  				int iCurHours = timeCurrent.getHours();
  				int iCurMin = timeCurrent.getMinutes();
  				int iMinHours = IaaDebug.timeCrawlStartTime.getHours();
  				int iMinMin = IaaDebug.timeCrawlStartTime.getMinutes();
  				int iMaxHours = IaaDebug.timeCrawlEndTime.getHours();
  				int iMaxMin = IaaDebug.timeCrawlEndTime.getMinutes();
  				boolean bValue = true;
  				if (iCurHours < iMinHours){
  					bValue = false;
  				} else if (iCurHours == iMinHours){
  					if ( iCurMin <= iMinMin ){
  						bValue = false;
  					}
  				} else if (iCurHours > iMaxHours){
  					bValue = false;
  				} else if (iCurHours == iMaxHours){
  					if ( iCurMin >= iMaxMin ){
  						bValue = false;
  					}
  				}
  				if ( m_bDispatch != bValue ){
					m_bDispatch = bValue;
  					log( "CrawlTimeThread has modify m_bDispatch, now it's value is " + m_bDispatch );
  				}
  				if ( !m_bDispatch ){
 					/* log every 10 minutes
 					 */
  					m_iMonitorTimes++;
  					if ( m_iMonitorTimes == 1 ){
  	  					log( "Now, because time is over to CrawlStartTime and CrawlEndTime, all request won't be dispatch." );
  					}
   					if ( m_iMonitorTimes > 600000/DISPATCH_MONITOR_FREQUENCY ) {
  						m_iMonitorTimes = 0;
  					}
  				} else {
  					if (m_iMonitorTimes != 0){
  						m_iMonitorTimes = 0;
  					}
  				}
  				try{
  					Thread.sleep(DISPATCH_MONITOR_FREQUENCY);
  				} catch (InterruptedException e){
  					//
  				}
  			}
  		}
  	};
  	m_objCrawlTimeThread.start();
  }
}
